Anweisung zur Durchführung der Datenfilterung und Weiterverarbeitung von Medien auf Tiktok

Work in Process

Tiktok Video-Link:

https://tiktok.com/@Kanalname/video/43664534324252233

Tiktok-Photo-Link:

https://toktok.com/@Kanalname/photo/436632342453252233

Im Rahmen unseres Projektes wurde eine umfassende Erhebung von nutzergenerierten Daten aus der Plattform TikTok durchgeführt. Diese Daten werden nun einer systematischen Aufbereitung und Filterung unterzogen, um sie für die anschließende Weiterverarbeitung vorzubereiten.

Ein wesentlicher Bestandteil dieses Prozesses ist die Identifikation deutscher Inhalte, welche durch speziell entwickelte Erkennungsverfahren erfolgt. Zudem findet eine manuelle Selektion statt, bei welcher durch den Einsatz einer Blacklist unerwünschte Kanäle ausgeschlossen werden.

Datenbank sichten

Alle Feststellungen sind akribisch zu dokumentiet worden. Die sqlite3 Datenbank wird zur weiteren Verarbeitung in ein Dataframe überführt.

# Datenbankinhalt in einen Pandas-DataFrame laden

import pandas as pd
import sqlite3
def db_to_dataframe(db_name="tiktok_media_metadata.db"):
    conn = sqlite3.connect(db_name)
    query = "SELECT * FROM media_metadata"
    try:
        df = pd.read_sql_query(query, conn)
        return df
    except sqlite3.Error as e:
        print(f"Fehler: {e}")
        return None
    finally:
        conn.close()

df = db_to_dataframe()

df

False-Positve-Filter

Die Erkennung deutscher Sprachinhalte erfolgt durch das erstmalige Bereinigen der zu prüfenden Texte von vorhandenen Hashtags. Es werden anhand von Wahrscheinlichkeiten das Vorhandensein der deutschen Sprache innerhalb gegebener Inhalte bestätigen oder ablehnen. Basierend auf den Ergebnissen wird der Datenbestand in besonders gekennzeichnete Textdateien geteilt, die dann zur weiteren Bearbeitung an die entsprechenden Stellen zugewiesen werden können.

import pandas as pd
from langdetect import detect_langs
import logging
from logging.handlers import RotatingFileHandler

# Loggings mit RotatingFileHandler (max. 10 MB)
logger = logging.getLogger("GermanDetection")
logger.setLevel(logging.DEBUG)
handler = RotatingFileHandler("process.log", maxBytes=10*1024*1024, backupCount=1)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

def is_german(text):
    try:
        # Entferne Hashtags, bevor die Sprache erkannt wird
        text_without_hashtags = ' '.join(word for word in text.split() if not word.startswith('#'))
        # Falls kein anderer Text nach dem Entfernen von Hashtags übrig bleibt
        if not text_without_hashtags.strip():
            return False

        # Erkennung der Sprache und Überprüfung der Wahrscheinlichkeit
        detected_languages = detect_langs(text_without_hashtags)
        for lang in detected_languages:
            if lang.lang == 'de' and lang.prob > 0.7:
                return True
        return False

    except Exception as e:
        logger.warning(f"Fehler beim Erkennen der Sprache: {e}")
        return False

logger.info("Starte Spracherkennung für DataFrame")

# Anwenden der Sprachprüfung auf Titel und Beschreibungen
df['german_title'] = df['title'].apply(is_german)
df['german_description'] = df['description'].apply(lambda x: is_german(x) if isinstance(x, str) else False)

logger.info("Spracherkennung abgeschlossen, gruppiere nach Uploader")

# Gruppierung nach Uploader und Aggregation der deutschen Inhalte
grouped = df.groupby('uploader').agg(
    total_entries=('title', 'count'),
    german_title_sum=('german_title', 'sum'),
    german_desc_sum=('german_description', 'sum')
)

grouped['german_average'] = (grouped['german_title_sum'] + grouped['german_desc_sum']) / (2 * grouped['total_entries'])

# Auswahl der Uploader die den deutschen Schwellenwert erfüllen
eligible_uploaders = grouped[grouped['german_average'] > 0.5].index.tolist()
logger.info(f"Anzahl der gefilterten Uploader: {len(eligible_uploaders)}")

# Ermittlung der Uploader die den Schwellenwert nicht erfüllen
all_uploaders = grouped.index.tolist()
non_eligible_uploaders = list(set(all_uploaders) - set(eligible_uploaders))
logger.info(f"Anzahl der nicht gefilterten Uploader: {len(non_eligible_uploaders)}")

# Filterung des ursprünglichen DataFrames nach den berechtigten Uploadern
result_df = df[df['uploader'].isin(eligible_uploaders)].copy()
logger.info(f"Ergebnis-DataFrame enthält {len(result_df)} Einträge")

# Ausgabe des gefilterten DataFrames in Jupyter Notebook
result_df

# Erstellen einer positives.txt mit detaillierten Informationen der gefilterten Einträge
with open("positives.txt", "w", encoding="utf-8") as f:
    for idx, row in result_df.iterrows():
        f.write(f"Uploader: {row['uploader']}\n")
        f.write(f"Title: {row['title']}\n")
        f.write(f"Description: {row.get('description', '')}\n")
        f.write("-" * 40 + "\n")

logger.info("positives.txt wurde erstellt.")

# Erstellen einer negatives.txt mit den unique Uploadern, die nicht die Kriterien erfüllen
with open("negatives.txt", "w", encoding="utf-8") as f:
    for uploader in non_eligible_uploaders:
        f.write(f"{uploader}\n")

logger.info("negatives.txt wurde erstellt.")

result_df

Tiktok-Kanal-Blacklist anwenden

Gemäß der geltenden Dienstvorschrift ist eine umfassende Überprüfung der Tiktok-Kanäle vorzunehmen, wobei die bestehende Blacklist mit erhöhter Sorgfalt anzuwenden ist.

import pandas as pd
# Blacklist einlesen
with open('blacklist.txt', 'r') as file:
    blacklist = [line.strip() for line in file if line.strip()]

# Filterung des DataFrames: Behalte nur die Einträge, deren Uploader NICHT in der Blacklist ist
filtered_result_df = result_df[~result_df['uploader'].isin(blacklist)]

# Anzeigen des gefilterten DataFrames
filtered_result_df

df = filtered_result_df

df to database

Den gefilterten DataFrame in einer neuen SQLite-Datenbank mit der gleichen Tabellenstruktur abspeichern.

import sqlite3

# Funktion zum Speichern des DataFrames in eine neue SQLite-Datenbank
def save_filtered_df_to_new_db(df, new_db_name="filtered_tiktok_media_metadata.db"):
    # Verbindung zur neuen Datenbank herstellen (erstellt sie, falls nicht vorhanden)
    conn = sqlite3.connect(new_db_name)

    # Sicherstellen, dass der DataFrame die Spalten in der gleichen Reihenfolge wie die Originaldatenbank hat
    try:
        # Die Tabelle in der neuen Datenbank erstellen und Daten einfügen
        df.to_sql('media_metadata', conn, if_exists='replace', index=False)
        print(f"Gefilterter DataFrame wurde erfolgreich in {new_db_name} gespeichert.")
    except sqlite3.Error as e:
        print(f"Fehler beim Speichern des DataFrames: {e}")
    finally:
        conn.close()

# Aufruf der Funktion, um den result_df zu speichern
save_filtered_df_to_new_db(df)

Gefilterter DataFrame wurde erfolgreich in filtered_tiktok_media_metadata.db gespeichert.

# Datenbankinhalt in einen Pandas-DataFrame laden

import pandas as pd
import sqlite3
def db_to_dataframe(db_name="filtered_tiktok_media_metadata.db"):
    conn = sqlite3.connect(db_name)
    query = "SELECT * FROM media_metadata"
    try:
        df = pd.read_sql_query(query, conn)
        return df
    except sqlite3.Error as e:
        print(f"Fehler: {e}")
        return None
    finally:
        conn.close()

df = db_to_dataframe()

df